package net.xuele.learn.mq;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public class MessageListener implements MessageListenerConcurrently {
    private MessageProcessor messageProcessor;

    public void setMessageProcessor(MessageProcessor messageProcessor) {
        this.messageProcessor = messageProcessor;
    }
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        try{
            for (MessageExt msg : list) {
                Boolean result = messageProcessor.handleMessage(msg);
                if (result == null) {
                    log.error("MessageListener consumeMessage result is null, msg Key : {}, msgId : {}, queueId : {}, QueueOffset :{}," +
                                    "delayLevel :{}",
                            msg.getKeys(), msg.getMsgId(), msg.getQueueId(), msg.getQueueOffset(), msg.getDelayTimeLevel());
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                if (!result) {
                    log.error("MessageListener consumeMessage result is error msg Key : {}, msgId : {}, queueId : {}, QueueOffset :{}," +
                                    "delayLevel :{}",
                            msg.getKeys(), msg.getMsgId(), msg.getQueueId(), msg.getQueueOffset(), msg.getDelayTimeLevel());
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            }
        } catch (Exception e) {
            log.error("error is {}", e);
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
